package com.aiwiown.snackmq.storage.segment;

import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.protocol.pb.SnackMessageProto;
import com.aiwiown.snackmq.common.serialization.Serializer;
import com.aiwiown.snackmq.common.serialization.SerializerFactory;
import com.aiwiown.snackmq.common.serialization.constants.SerializerAlgorithm;
import com.aiwiown.snackmq.common.util.MessageConverter;
import com.aiwiown.snackmq.storage.index.Index;
import com.aiwiown.snackmq.storage.index.IndexEntry;
import com.aiwiown.snackmq.storage.util.BufferCleaner;
import com.aiwiown.snackmq.storage.util.ResourceManager;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Getter
public class LogSegment {

    private final File logFile;
    private final RandomAccessFile randomAccessFile;
    private FileChannel logChannel;
    private MappedByteBuffer mappedLogBuffer;
    private final long baseOffset;
    private final long segmentSize;
    private final int indexInterval;
    private final Serializer serializer;
    private Index index;
    private AtomicInteger logPosition;
    private final AtomicInteger entryCount = new AtomicInteger(0);
    // 【新增】用于在恢复时精确追踪段内最高的逻辑位点
    private final AtomicLong maxLogicalOffset = new AtomicLong(-1L);
    private final AtomicBoolean closed = new AtomicBoolean(false);
    // 【新增】恢复策略标志
    private final boolean forceFullRecovery;


    public LogSegment(String basePath, long baseOffset, long segmentSize, int indexInterval, boolean forceFullRecovery) throws IOException {
        this(basePath, baseOffset, segmentSize, indexInterval, forceFullRecovery, SerializerFactory.getSerializer(SerializerAlgorithm.PROTOBUF));
    }

    /**
     * 【核心性能修复】重构构造函数，采用基于索引的快速恢复机制。
     * 旧的实现会完全重扫描日志文件来重建索引，导致在有大量消息时启动非常缓慢。
     * 新的实现首先加载现有索引，然后只从最后一个索引点开始扫描日志文件的尾部，
     * 极大地加快了 Broker 的启动速度。
     */
    public LogSegment(String basePath, long baseOffset, long segmentSize, int indexInterval, boolean forceFullRecovery, Serializer serializer) throws IOException {
        // 【新增】确保目录存在
        File logDir = new File(basePath);
        if (!logDir.exists()) {
            boolean created = logDir.mkdirs();
            if (!created) {
                log.error("Failed to create log segment directory: {}", basePath);
            }
        }
        this.baseOffset = baseOffset;
        this.segmentSize = segmentSize;
        this.indexInterval = indexInterval;
        this.serializer = serializer;
        this.forceFullRecovery = forceFullRecovery;

        final String baseFileName = String.format("%020d", baseOffset);
        this.logFile = new File(basePath, baseFileName + ".log");
        final File indexFile = new File(basePath, baseFileName + ".index");
        // 1. 【最终修复】打开并保存 RandomAccessFile 实例
        this.randomAccessFile = new RandomAccessFile(logFile, "rw");
        this.logChannel = this.randomAccessFile.getChannel();

        long maxIndexSize = (segmentSize / (long) indexInterval / 8) * IndexEntry.SIZE;
        if (maxIndexSize <= 0 && segmentSize > 0) {
            maxIndexSize = 1024 * 1024;
        }
        this.index = new Index(indexFile, maxIndexSize);
        // 2. 基于索引进行快速恢复
        // 【修改】调用新的 recover 方法，它现在会返回一个包含更多信息的结果对象
        RecoveryResult recoveryResult = recover();
        this.logPosition = new AtomicInteger(recoveryResult.physicalPosition);
        this.entryCount.set(recoveryResult.entryCount);
        // 【新增】在恢复后，设置精确的最高逻辑位点
        this.maxLogicalOffset.set(recoveryResult.maxLogicalOffset);

        // 3. 【修复】在恢复和截断完成后再进行内存映射，确保映射的是一个干净的文件视图
        // 这样可以避免在映射后无法截断文件的问题
        // 注意：这里不再自动映射，而是在需要时再映射
        log.info("从段 {} 恢复完成。最后有效物理位置: {}, 有效条目数: {}, 最高逻辑位点: {}",
                this.baseOffset, this.logPosition.get(), this.entryCount.get(), this.maxLogicalOffset.get());
    }

    @Getter
    @AllArgsConstructor
    private static class RecoveryResult {
        private final int physicalPosition;
        private final int entryCount;
        private final long maxLogicalOffset;
    }

    /**
     * 【新增】修复损坏的索引文件。
     * 当索引文件损坏时，重新扫描日志文件并重建索引。
     *
     * @throws IOException 如果修复失败
     */
    public synchronized void repairIndex() throws IOException {
        log.info("开始修复段 {} 的索引文件", this.baseOffset);

        // 清空当前索引
        this.index.truncateAndClear();

        // 重新扫描整个日志文件
        RecoveryResult result = scanAndRebuild(0, 0, -1L);
        this.logPosition.set(result.getPhysicalPosition());
        this.entryCount.set(result.getEntryCount());
        this.maxLogicalOffset.set(result.getMaxLogicalOffset());


        log.info("段 {} 索引修复完成。最后有效物理位置: {}, 有效条目数: {}, 最高逻辑位点: {}",
                this.baseOffset, this.logPosition.get(), this.entryCount.get(), this.maxLogicalOffset.get());
    }

    /**
     * 【新增】检查索引是否损坏。
     *
     * @return 如果索引损坏返回true
     */
    public boolean isIndexCorrupted() {
        try {
            IndexEntry lastEntry = this.index.getLastEntry();
            if (lastEntry == null) {
                return false; // 空索引不算损坏
            }

            // 检查最后一个条目的有效性
            if (lastEntry.getOffset() < this.baseOffset) {
                return true;
            }

            // 由于并发写入可能导致位点不连续，基于条目数的检查不再可靠，因此移除。
            // long expectedEntries = lastEntry.getOffset() - this.baseOffset + 1;
            // if (expectedEntries != this.entryCount.get()) {
            //     return true;
            // }

            return false;
        } catch (Exception e) {
            log.warn("检查索引损坏时发生异常", e);
            return true;
        }
    }

    /**
     * 【最终修复】基于索引的快速恢复方法。
     * 此方法现在会精确计算扫描的起始点，并调用新的 scanAndRebuild 方法来
     * 找出段内真正的最高逻辑位点，以应对并发写入导致的位点乱序问题。
     *
     * @return 包含恢复后状态的 RecoveryResult 对象。
     * @throws IOException 如果 I/O 操作失败。
     */
    private RecoveryResult recover() throws IOException {
        if (this.forceFullRecovery) {
            // 【安全策略】为了保证在所有情况下的数据恢复正确性，执行全量扫描。
            // 全量扫描虽然在启动时理论上较慢，但它保证了状态的绝对正确，这对于存储系统至关重要。
            log.info("【全量恢复】根据配置，将对段 {} 执行全量扫描恢复。", this.baseOffset);
            this.index.load(); // 仍然加载索引，但主要目的是为了后续的重建
            this.index.truncateAndClear(); // 清空旧索引，因为 scanAndRebuild 会重建它
            return scanAndRebuild(0, 0, -1L);
        } else {
            // 【性能策略】基于索引进行快速恢复
            this.index.load();
            IndexEntry lastIndexedEntry = this.index.getLastEntry();
            if (lastIndexedEntry != null) {
                log.info("【快速恢复】段 {} 发现索引，将从物理位置 {} 开始增量恢复。", this.baseOffset, lastIndexedEntry.getPosition());
                return scanAndRebuild(lastIndexedEntry.getPosition(), this.index.getEntryCount(), lastIndexedEntry.getOffset());
            } else {
                log.info("【快速恢复-回退】段 {} 未发现有效索引，将回退到全量扫描恢复。", this.baseOffset);
                return scanAndRebuild(0, 0, -1L);
            }
        }
    }


    /**
     * 【修改】从指定位置扫描日志文件，重建状态并截断损坏的数据。
     * 现在此方法会解析消息以找到真正的最高逻辑位点。
     *
     * @param startPosition  物理扫描的起始位置。
     * @param knownEntries   在起始位置之前已知的有效条目数。
     * @param knownMaxOffset 在起始位置之前已知的最高逻辑位点。
     * @return 包含最终物理位置、条目数和最高逻辑位点的 RecoveryResult。
     * @throws IOException 如果 I/O 操作失败。
     */
    private RecoveryResult scanAndRebuild(int startPosition, int knownEntries, long knownMaxOffset) throws IOException {
        long fileSize = this.logChannel.size();
        if (fileSize == 0) {
            return new RecoveryResult(0, 0, -1L);
        }
        // 使用内存映射文件进行高效扫描
        MappedByteBuffer scanBuffer = this.logChannel.map(FileChannel.MapMode.READ_ONLY, 0, fileSize);

        int currentPosition = startPosition;
        int currentEntryCount = knownEntries;
        long currentMaxLogicalOffset = knownMaxOffset;

        try {
            ((Buffer) scanBuffer).position(startPosition);

            while (scanBuffer.hasRemaining()) {
                if (scanBuffer.remaining() < 4) {
                    log.warn("在段 {} 的位置 {} 发现不完整的记录头。正在截断。", this.baseOffset, scanBuffer.position());
                    break;
                }
                int recordStartPosition = scanBuffer.position();
                int recordSize = scanBuffer.getInt();

                if (recordSize <= 0 || scanBuffer.remaining() < recordSize) {
                    log.warn("在段 {} 的位置 {} 发现无效或不完整的记录。正在截断。", this.baseOffset, recordStartPosition);
                    scanBuffer.position(recordStartPosition);
                    break;
                }

                int messagePosition = scanBuffer.position();
                byte[] messageBytes = new byte[recordSize];
                scanBuffer.get(messageBytes);

                long logicalOffset = -1L;

                try {
                    SnackMessageProto.SnackMessage protoMessage = this.serializer.deserialize(messageBytes, SnackMessageProto.SnackMessage.class);
                    String offsetStr = protoMessage.getHeader().getPropertiesMap().get("offset");
                    if (offsetStr != null) {
                        logicalOffset = Long.parseLong(offsetStr);
                        currentMaxLogicalOffset = Math.max(currentMaxLogicalOffset, logicalOffset);
                    }
                } catch (Exception e) {
                    log.warn("无法反序列化或解析位点，在位置 {} 的记录可能已损坏。截断。", messagePosition, e);
                    scanBuffer.position(recordStartPosition);
                    break;
                }

                // 【修复】只为增量扫描到的部分重建索引
                if (logicalOffset != -1L && logicalOffset > knownMaxOffset && logicalOffset % this.indexInterval == 0) {
                    if (!this.index.isFull()) {
                        this.index.append(logicalOffset, recordStartPosition);
                    }
                }

                currentPosition = scanBuffer.position();
                currentEntryCount++;
            }
        } finally {
            BufferCleaner.clean(scanBuffer);
        }

        if (currentPosition < fileSize) {
            log.info("正在将日志文件 {} 从大小 {} 截断到有效位置 {}", this.logFile.getName(), fileSize, currentPosition);
            try {
                // 【修复Windows平台Bug】在截断前，必须先释放内存映射
                if (this.mappedLogBuffer != null) {
                    BufferCleaner.clean(this.mappedLogBuffer);
                    this.mappedLogBuffer = null;
                }
                this.logChannel.truncate(currentPosition);
            } catch (IOException e) {
                log.error("截断日志文件失败，这在Windows上可能是因为文件仍被映射。", e);
                // 即使截断失败，也要继续，因为状态是正确的
            }
        }
        return new RecoveryResult(currentPosition, currentEntryCount, currentMaxLogicalOffset);
    }
    /**
     * 【修改】追加一条消息到日志文件中。
     *
     * @param message       要追加的消息。
     * @param logicalOffset 消息的逻辑偏移量。
     * @return 追加的消息大小。
     * @throws IOException 如果 I/O 操作失败。
     */
    public synchronized int append(Message message, long logicalOffset) throws IOException {
        if (closed.get()) {
            throw new IOException("Segment " + baseOffset + " is closed.");
        }

        ensureMapped();

        // 【新增】在追加时，也更新 maxLogicalOffset
        this.maxLogicalOffset.getAndAccumulate(logicalOffset, Math::max);

        // 【修复】在序列化之前，将 logicalOffset 添加到消息属性中
        // 1. 将业务 Message 对象转换为 Protobuf Message 对象
        SnackMessageProto.SnackMessage originalProto = MessageConverter.toProto(message);
        // 2. 创建一个 Builder 以便修改
        SnackMessageProto.SnackMessage.Builder protoBuilder = originalProto.toBuilder();

        // 3. 获取 Header 的 Builder。如果 Header 不存在，则创建一个新的。
        SnackMessageProto.SnackMessage.Header.Builder headerBuilder = protoBuilder.hasHeader()
                ? protoBuilder.getHeader().toBuilder()
                : SnackMessageProto.SnackMessage.Header.newBuilder();

        // 4. 将 offset 属性放入 Header 的 properties map 中
        headerBuilder.putProperties("offset", String.valueOf(logicalOffset));

        // 5. 将修改后的 Header 设置回消息 Builder
        protoBuilder.setHeader(headerBuilder);

        // 6. 构建最终的、包含 offset 的 Proto 消息
        SnackMessageProto.SnackMessage finalProtoMessage = protoBuilder.build();

        // 7. 序列化最终的消息
        byte[] messageBytes = this.serializer.serialize(finalProtoMessage);
        int recordSize = 4 + messageBytes.length;
        if (isFull(recordSize)) {
            throw new SegmentFullException("段已满，无法追加 " + recordSize + " 字节的数据。");
        }

        int currentLogPos = logPosition.get();

        // 使用逻辑位点 (logicalOffset) 而不是内部条目计数 (entryCount) 来决定是否创建索引。
        if (logicalOffset % indexInterval == 0) {
            index.append(logicalOffset, currentLogPos);
        }

        entryCount.getAndIncrement();

        // 强制类型转换为 Buffer 以兼容 JDK 8
        ((Buffer) mappedLogBuffer).position(currentLogPos);
        mappedLogBuffer.putInt(messageBytes.length);
        mappedLogBuffer.put(messageBytes);
        logPosition.addAndGet(recordSize);
        return recordSize;
    }


    /**
     * 【新增】真正高效的批量追加方法。
     *
     * @param messages        要追加的消息列表。
     * @param startOffset     此批次消息的起始逻辑位点。
     * @param commitLogOffset 此批次在 Raft 日志中的物理位点。
     * @param partitionId     此段所属的分区ID。
     * @return 成功写入此段的消息数量。
     * @throws IOException          如果写入日志文件或索引文件失败。
     * @throws SegmentFullException 如果当前段已满，连批次中的第一条消息都无法容纳。
     */
    public synchronized int appendBatch(List<Message> messages, long startOffset, long commitLogOffset, int partitionId) throws IOException, SegmentFullException {
        if (closed.get()) {
            throw new IOException("Segment " + baseOffset + " is closed.");
        }
        if (messages == null || messages.isEmpty()) {
            return 0;
        }

        ensureMapped();

        int messagesWritten = 0;
        int bytesWritten = 0;
        int currentPositionInSegment = this.logPosition.get();
        final Map<String, String> extraProperties = new HashMap<>(3);
        extraProperties.put(Message.PROPERTY_PHYSICAL_OFFSET, String.valueOf(commitLogOffset));
        extraProperties.put("partition", String.valueOf(partitionId));

        for (int i = 0; i < messages.size(); i++) {
            Message msg = messages.get(i);
            long currentLogicalOffset = startOffset + i;
            extraProperties.put("offset", String.valueOf(currentLogicalOffset));
            byte[] msgBytes = MessageConverter.toBytes(msg, extraProperties);
            if (msgBytes == null) {
                throw new IOException("Failed to serialize message with extra properties: " + msg);
            }
            int recordSize = 4 + msgBytes.length;
            if (currentPositionInSegment + bytesWritten + recordSize > this.segmentSize || index.isFull()) {
                break;
            }
            // 直接写入 MappedByteBuffer
            ((Buffer) mappedLogBuffer).position(currentPositionInSegment + bytesWritten);
            mappedLogBuffer.putInt(msgBytes.length);
            mappedLogBuffer.put(msgBytes);

            // 如果需要，更新索引
            if (currentLogicalOffset % this.indexInterval == 0) {
                this.index.append(currentLogicalOffset, currentPositionInSegment + bytesWritten);
            }
            bytesWritten += recordSize;
            messagesWritten++;
        }
        if (messagesWritten == 0 && !messages.isEmpty()) {
            throw new SegmentFullException("Segment is full, cannot even fit the first message of the batch.");
        }
        this.logPosition.addAndGet(bytesWritten);
        this.entryCount.addAndGet(messagesWritten);
        // 更新最高位点
        if (messagesWritten > 0) {
            this.maxLogicalOffset.getAndAccumulate(startOffset + messagesWritten - 1, Math::max);
        }
        return messagesWritten;
    }

    /**
     * 【最终修复】读取一条消息。此方法现在是线程安全的，并且能够处理并发写入导致的乱序消息。
     * 它会从索引点开始完整扫描，直到找到目标消息或扫描完所有记录。
     */
    public synchronized Message read(long targetOffset) throws IOException {
        // 1. 边界检查
        if (targetOffset < baseOffset || targetOffset > this.maxLogicalOffset.get()) {
            return null;
        }

        // 2. 确保文件已映射
        ensureMapped();

        // 3. 通过索引查找起始扫描点
        IndexEntry entry = index.lookup(targetOffset);
        if (entry == null) {
            log.warn("在段 {} 中未找到偏移量 {} 的任何有效索引起始点。", baseOffset, targetOffset);
            // 【修复】即使索引未找到，也从头开始扫描，以应对索引尚未写入的场景
            entry = new IndexEntry(baseOffset, 0);
        }

        // 4. 创建一个线程安全的、拥有独立 position 的 buffer 视图用于扫描
        ByteBuffer logView = ((ByteBuffer) mappedLogBuffer).duplicate();
        // 【最终修复】将扫描范围限制在已写入的数据区域内
        ((Buffer) logView).limit(this.logPosition.get());
        ((Buffer) logView).position(entry.getPosition());

        // 5. 【修复】从索引点开始扫描，直到文件末尾，不再依赖 currentScannedOffset
        while (logView.hasRemaining()) {
            if (logView.remaining() < 4) break; // 记录头不完整

            int recordStartPosition = logView.position();
            int messageLength = logView.getInt();

            if (messageLength <= 0 || logView.remaining() < messageLength) {
                log.warn("在段 {} 的物理位置 {} 发现无效或损坏的记录，读取中止。", baseOffset, recordStartPosition);
                break; // 记录损坏
            }

            int messageBodyPosition = logView.position();
            byte[] messageBytes = new byte[messageLength];
            // 使用一个临时的 duplicate 来读取，这样不会影响主 logView 的 position
            ByteBuffer tempView = logView.duplicate();
            tempView.get(messageBytes);

            try {
                SnackMessageProto.SnackMessage protoMessage = this.serializer.deserialize(messageBytes, SnackMessageProto.SnackMessage.class);
                String offsetStr = protoMessage.getHeader().getPropertiesMap().get("offset");
                if (offsetStr != null) {
                    long actualOffset = Long.parseLong(offsetStr);
                    if (actualOffset == targetOffset) {
                        // 找到了！
                        return MessageConverter.fromProto(protoMessage);
                    }
                }
            } catch (Exception e) {
                log.warn("读取或反序列化消息时出错，在物理位置 {}。跳过。", messageBodyPosition, e);
            }

            // 将主扫描视图的位置移动到下一条记录的开始
            ((Buffer) logView).position(messageBodyPosition + messageLength);
        }

        return null; // 扫描完整个段后未找到
    }

    public boolean isFull() {
        return logPosition.get() >= segmentSize || index.isFull();
    }

    private boolean isFull(int requiredSpace) {
        return logPosition.get() + requiredSpace > segmentSize || index.isFull();
    }

    public void flush() {
        if (closed.get()) {
            return;
        }
        if (mappedLogBuffer != null) {
            mappedLogBuffer.force();
        }
        if (index != null) {
            index.flush();
        }
    }

    public void close() throws IOException {
        if (!closed.compareAndSet(false, true)) {
            return;
        }

        log.debug("正在关闭段 {}", baseOffset);
        try {
            flush();
        } finally {
            if (mappedLogBuffer != null) {
                BufferCleaner.clean(this.mappedLogBuffer);
                this.mappedLogBuffer = null;
            }
            if (index != null) {
                index.close();
                this.index = null;
            }
            if (randomAccessFile != null) {
                randomAccessFile.close();
            }
            this.logChannel = null;
            ResourceManager.forceGcAndWait(100);
            log.debug("段 {} 关闭完成。", baseOffset);
        }
    }

    public boolean isClosed() {
        return this.closed.get();
    }

    /**
     * 【最终修复】零拷贝读取方法 - 直接返回ByteBuffer切片，避免内存拷贝。
     * 此方法现在能够处理并发写入导致的乱序消息。
     */
    public synchronized ByteBuffer readZeroCopy(long targetOffset) throws IOException {
        // 1. 边界检查
        if (targetOffset < baseOffset || targetOffset > this.maxLogicalOffset.get()) {
            return null;
        }

        // 2. 确保文件已映射到内存
        ensureMapped();

        // 3. 通过索引查找起始扫描点
        IndexEntry entry = index.lookup(targetOffset);
        if (entry == null) {
            log.warn("在 readZeroCopy 中，段 {} 未找到位点 {} 的任何有效索引起始点。将从头扫描。", baseOffset, targetOffset);
            entry = new IndexEntry(baseOffset, 0);
        }

        // 4. 创建一个线程安全的、拥有独立 position 的 buffer 视图用于扫描
        ByteBuffer logView = ((ByteBuffer) mappedLogBuffer).duplicate();
        // 【最终修复】将扫描范围限制在已写入的数据区域内
        ((Buffer) logView).limit(this.logPosition.get());
        ((Buffer) logView).position(entry.getPosition());

        // 5. 【修复】从索引点开始扫描，直到文件末尾
        while (logView.hasRemaining()) {
            if (logView.remaining() < 4) break; // 记录头不完整

            int recordStartPosition = logView.position();
            int messageLength = logView.getInt();

            if (messageLength <= 0 || logView.remaining() < messageLength) {
                log.warn("在段 {} 的物理位置 {} 发现无效或损坏的记录，读取中止。", baseOffset, recordStartPosition);
                break; // 记录损坏
            }

            int messageBodyPosition = logView.position();
            byte[] messageBytes = new byte[messageLength];
            // 使用一个临时的 duplicate 来读取，这样不会影响主 logView 的 position
            ByteBuffer tempView = logView.duplicate();
            tempView.get(messageBytes);

            try {
                SnackMessageProto.SnackMessage protoMessage = this.serializer.deserialize(messageBytes, SnackMessageProto.SnackMessage.class);
                String offsetStr = protoMessage.getHeader().getPropertiesMap().get("offset");
                if (offsetStr != null) {
                    long actualOffset = Long.parseLong(offsetStr);
                    if (actualOffset == targetOffset) {
                        // 找到了！现在创建零拷贝切片并返回。
                        ByteBuffer resultBuffer = ((ByteBuffer) mappedLogBuffer).duplicate();
                        ((Buffer) resultBuffer).position(messageBodyPosition);
                        ((Buffer) resultBuffer).limit(messageBodyPosition + messageLength);
                        return resultBuffer.slice(); // slice() 创建一个共享内容的 buffer
                    }
                }
            } catch (Exception e) {
                log.warn("在零拷贝读取期间，反序列化物理位置 {} 的消息时出错。跳过。", messageBodyPosition, e);
            }

            // 将主扫描视图的位置移动到下一条记录的开始
            ((Buffer) logView).position(messageBodyPosition + messageLength);
        }

        return null; // 未找到
    }

    /**
     * 【最终修复】批量零拷贝读取方法 - 一次性读取多个消息。
     * 此方法现在能够处理并发写入导致的乱序消息。
     */
    public synchronized List<ByteBuffer> readBatchZeroCopy(long startOffset, int maxMessages) throws IOException {
        List<ByteBuffer> results = new ArrayList<>();
        if (maxMessages <= 0) {
            return results;
        }

        // 1. 边界检查
        if (startOffset < baseOffset || startOffset > this.maxLogicalOffset.get()) {
            return results;
        }

        // 2. 确保文件已映射
        ensureMapped();

        // 3. 查找起始点
        IndexEntry entry = index.lookup(startOffset);
        if (entry == null) {
            log.warn("在 readBatchZeroCopy 中，段 {} 未找到起始位点 {} 的任何有效索引起始点。将从头扫描。", baseOffset, startOffset);
            entry = new IndexEntry(baseOffset, 0);
        }

        // 4. 创建扫描视图
        ByteBuffer logView = ((ByteBuffer) mappedLogBuffer).duplicate();
        // 【最终修复】将扫描范围限制在已写入的数据区域内
        ((Buffer) logView).limit(this.logPosition.get());
        ((Buffer) logView).position(entry.getPosition());

        // 5. 【修复】从索引点开始扫描，直到读取足够的消息或到达文件末尾
        while (results.size() < maxMessages && logView.hasRemaining()) {
            if (logView.remaining() < 4) break;

            int recordStartPosition = logView.position();
            int messageLength = logView.getInt();

            if (messageLength <= 0 || logView.remaining() < messageLength) {
                log.warn("在段 {} 的物理位置 {} 发现无效或损坏的记录，批量读取中止。", baseOffset, recordStartPosition);
                break;
            }

            int messageBodyPosition = logView.position();
            byte[] messageBytes = new byte[messageLength];
            ByteBuffer tempView = logView.duplicate();
            tempView.get(messageBytes);

            try {
                SnackMessageProto.SnackMessage protoMessage = this.serializer.deserialize(messageBytes, SnackMessageProto.SnackMessage.class);
                String offsetStr = protoMessage.getHeader().getPropertiesMap().get("offset");
                if (offsetStr != null) {
                    long actualOffset = Long.parseLong(offsetStr);
                    if (actualOffset >= startOffset) {
                        // 这条消息是批次的一部分，创建切片
                        ByteBuffer resultBuffer = ((ByteBuffer) mappedLogBuffer).duplicate();
                        ((Buffer) resultBuffer).position(messageBodyPosition);
                        ((Buffer) resultBuffer).limit(messageBodyPosition + messageLength);
                        results.add(resultBuffer.slice());
                    }
                }
            } catch (Exception e) {
                log.warn("在零拷贝批量读取期间，反序列化物理位置 {} 的消息时出错。跳过。", messageBodyPosition, e);
            }

            // 将主扫描视图的位置移动到下一条记录的开始
            ((Buffer) logView).position(messageBodyPosition + messageLength);
        }

        return results;
    }

    /**
     * 延迟映射文件，在需要时再创建内存映射
     */
    private void ensureMapped() throws IOException {
        if (this.mappedLogBuffer == null) {
            this.mappedLogBuffer = logChannel.map(FileChannel.MapMode.READ_WRITE, 0, segmentSize);
        }
    }
}